set hive.exec.dynamic.partition=true;
set hive.exec.dynamic.partition.mode=nonstrict;
set hive.exec.max.dynamic.partitions=1000;
set hive.exec.max.dynamic.partitions.pernode=1000;
with t_center_arrival as (
    select * from 
    (select 
    * ,
    row_number() over (partition by billcode order by scantime) as rnk
    from    
    (select     
    billcode,
    scansitecode,
    scantime,
    weight,
    remark,
    mobile,
    remark1,
    if(nvl(weight,0)<>0,1,0) as center_weightornot,
    date(scantime) as scan_dt,
    row_number() over (partition by billcode,scansitecode order by weight desc) as rk
    from jms_ods.tab_barscan_centerarrival
    where dt between date_format(date_add('{{ execution_date | cst_ds }}', -3), 'yyyy-MM-dd') and '{{ execution_date | cst_ds }}'
    and packagecode is null
    and substr(billcode, 1, 2) = 'JT'
    ) a
    where rk = 1    
    ) b
    where rnk = 1 and scan_dt = '{{ execution_date | cst_ds }}'
),
t_waybill as (
         select       way.waybill_no,
                      net.id  as  pick_network_id,
                      way.pick_network_code,
                      way.package_receipt_weight,
                      way.package_collect_weight,
                      way.package_arrival_weight,
                      way.package_volume
         from
             ( select waybill_no,
                      pick_network_code,
                      package_receipt_weight,  --入仓重量
                      package_collect_weight,  --收件重量
                      package_arrival_weight,  --集散到件重量
                      package_volume
               from jms_dwd.dwd_yl_oms_oms_waybill_dt
               where dt >= date_add('{{ execution_date | cst_ds }}', -3) and dt <= '{{ execution_date | cst_ds }}'
                 ) way
                 left join
             (select id,code from jms_dim.dim_lmdm_sys_network ) net
             on way.pick_network_code=code
     ),
     t_loading as (
         select
             billcode,
             scansitecode_early as scansitecode,
             nextstationcode_early as nextstationcode
         from jms_dwd.dwd_barscan_loading_dt
         where dt >= date_add('{{ execution_date | cst_ds }}', -3)
           and dt <= '{{ execution_date | cst_ds }}'
           and dimtype = 'bill_scansite'
     ),
     t_network as (
          select *
                  from jms_dim.dim_network_whole_massage
                  where is_delete = 1
                    and is_enable = 1
     ),
     t_network_distributi as (
         select b.network_id,
                b.id   as first_id,
                b.code as first_code,
                b.name as first_name,
                c.id   as second_id,
                c.code as second_code,
                c.name as second_name
         from jms_dim.dim_lmdm_sys_network_distributi b
                  left join jms_dim.dim_lmdm_sys_network_distributi c
                            on b.id = c.network_id
                                and c.distribution_type = 2         --可交货分拨
     ),
     t_network_path as (
         select id, code, name, agent_id as agent_id --代理区id
                 ,  fran_id as franchisee_id   --加盟商id
                 --path是树路径，/22/770/2232/4939/
                 from t_network
     ),
     t_network_final as (
         select t_network_path.id
              , t_network_path.code
              , t_network_path.name
              , t_network_agent.id        as agent_id
              , t_network_agent.code      as agent_code
              , t_network_agent.name      as agent_name
              , t_network_franchisee.id   as franchisee_id
              , t_network_franchisee.code as franchisee_code
              , t_network_franchisee.name as franchisee_name
         from t_network_path t_network_path
                          left join t_network t_network_agent
                                    on t_network_path.agent_id = t_network_agent.id and t_network_agent.network_type = 2
                          left join t_network t_network_franchisee
                                    on t_network_path.franchisee_id = t_network_franchisee.id and
                                       t_network_franchisee.network_type = 3
     )
insert overwrite table jms_dm.dm_cn_export_diff_weight_detail_dt partition (dt = '{{ execution_date | cst_ds }}')
select scan_date ,
       pick_network_id ,
       pick_network_code ,  
       pick_network_name,
       center_code,
       center_name,
       nodal_network_code,
       nodal_network_name, 
       pick_nk.agent_code as subordinate_agent_code,
       pick_nk.agent_name as subordinate_agent_name,
       pick_nk.franchisee_code as franchisee_code,
       pick_nk.franchisee_name as franchisee_name,
       billcode,
       compare_type,
       cast(compare_weight as decimal(13,2)),
       cast(center_weight as decimal(13,2)),
       cast((center_weight - compare_weight) as decimal(13,2)) as center_diff_weight,
       center_weightornot
from (select t_center_arrival.scan_dt                                        as                                                       scan_date,
             t_waybill.pick_network_code                                     as                                                       pick_network_code,
             t_waybill.pick_network_id                                       as                                                       pick_network_id,
             pick_nk.name                                                    as                                                       pick_network_name,
             t_center_arrival.scansitecode                                   as                                                       center_code,
             center_nk.name                                                  as                                                       center_name,
             case
                 when loading_nk.name like '%集散点' then t_loading.scansitecode
                 else ''
                 end                                                         as                                                       nodal_network_code,
             case
                 when loading_nk.name like '%集散点' then loading_nk.name
                 else ''
                 end                                                         as                                                       nodal_network_name,
             t_center_arrival.billcode,
             case
                 when nvl(t_waybill.package_receipt_weight, 0) >= nvl(t_waybill.package_collect_weight, 0) and
                      nvl(t_waybill.package_receipt_weight, 0) >= nvl(t_waybill.package_arrival_weight, 0)
                     then '入仓重量'
                 when nvl(t_waybill.package_collect_weight, 0) >= nvl(t_waybill.package_receipt_weight, 0) and
                      nvl(t_waybill.package_collect_weight, 0) >= nvl(t_waybill.package_arrival_weight, 0)
                     then '收件重量'
                 when nvl(t_waybill.package_arrival_weight, 0) >= nvl(t_waybill.package_receipt_weight, 0) and
                      nvl(t_waybill.package_arrival_weight, 0) >= nvl(t_waybill.package_collect_weight, 0)
                     then '集散到件重量'
                 end                                                         as                                                       compare_type,
             case
                 when t_center_arrival.weight <= 3.2 and
                      (nvl(t_center_arrival.remark, '0') <= 80 or nvl(t_center_arrival.mobile, '0') <= 80 or
                       nvl(t_center_arrival.remark1, '0') <= 80)
                     and (nvl(t_center_arrival.remark, '0') + nvl(t_center_arrival.mobile, '0') +
                          nvl(t_center_arrival.remark1, '0')) <= 100
                     then t_center_arrival.weight
                 else greatest(t_center_arrival.weight, nvl(t_waybill.package_volume, 0))
                 end                                                         as                                                       center_weight,
             greatest(cast(nvl(t_waybill.package_receipt_weight, 0) as decimal(13,2)), cast(nvl(t_waybill.package_collect_weight, 0) as decimal(13,2)),
                      cast(nvl(t_waybill.package_arrival_weight, 0) as decimal(13,2))) as                                                       compare_weight,
             center_weightornot,
             row_number() over (partition by t_waybill.pick_network_code,t_center_arrival.billcode order by t_center_arrival.scan_dt) rnk
      from t_center_arrival t_center_arrival
               join t_waybill t_waybill
                    on t_waybill.waybill_no = t_center_arrival.billcode
               left join t_loading t_loading
                         on t_center_arrival.billcode = t_loading.billcode and
                            t_center_arrival.scansitecode = t_loading.nextstationcode
               left join t_network pick_nk on pick_nk.code = t_waybill.pick_network_code
               left join t_network center_nk on center_nk.code = t_center_arrival.scansitecode
               left join t_network loading_nk on loading_nk.code = t_loading.scansitecode
     ) a
         left join t_network_final pick_nk on pick_nk.code = a.pick_network_code
         left semi
         join t_network_distributi t_network_distributi
              on a.pick_network_id = t_network_distributi.network_id
                  and (a.center_code = t_network_distributi.first_code or a.center_code = t_network_distributi.second_code)
where a.rnk = 1
DISTRIBUTE BY abs(hash(billcode)) % 4 ; 